package com.lq.esstudy.service.impl;


import com.lq.esstudy.entity.AriMysqlEs;
import com.lq.esstudy.service.EsDataService;
import com.lq.esstudy.utils.EsUtils;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;


@Service
@Slf4j
public class EsDataServiceImpl implements EsDataService {

    private static final String INIT_TIME = "1970-01-01 00:00:00";

    private static final int ES_COUNT = 300;

    @Resource
    private EsUtils esUtils;

    @Value("${dwd.ari_news}")
    private String index_ari_news;


    @Value("${nginx.url}")
    private String nginxUrl;

    @PostConstruct
    public void updateReadCountInit() {
        this.updateReadCountTask();
    }

    @Scheduled(cron = "${time.mysql_to_es}")
    public void updateEsByTime() {
        log.info("***开始更新mysql数据到es");
        //每天凌晨1点更新mysql数据到es
        this.mysqlToEs("");
        log.info("***完成更新mysql数据到es");
    }


    // 定时任务更新
    @Scheduled(cron = "${time.read_count:0 */5 * * * ?}")
    public void updateReadCountTask() {
        log.info("***开始更新阅读次数到es");
        //每隔5分钟更新阅读次数数据
        this.mysqlToEs(index_ari_news);
        log.info("***完成更新阅读次数到es");
    }


//    public IPage<Object> searchAriNews(DwdEsVO dwdEsVO) {
//        SearchSourceBuilder build = new SearchSourceBuilder();
//        //拿到输入框的内容
//        String search = dwdEsVO.getKeyword();
//
//        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
//        //高亮显示
//        HighlightBuilder highlightBuilder = new HighlightBuilder();
//        highlightBuilder.preTags("<font style='color:#57a4ea'>");
//        highlightBuilder.postTags("</font>");
//        //判断搜索是按全部搜索还是按某个类型搜索
//        if (StringUtils.isNotEmpty(search)) {
//            boolQueryBuilder.must(QueryBuilders.boolQuery()
//                    .should(QueryBuilders.matchPhraseQuery("title", search)).boost(10)
//                    .should(QueryBuilders.matchPhraseQuery("subtitle", search)).boost(10)
//                    .should(QueryBuilders.matchPhraseQuery("author", search)).boost(10)
//                    .should(QueryBuilders.matchPhraseQuery("keyword", search)).boost(10)
//                    .should(QueryBuilders.matchQuery("title", search)).boost(5)
//                    .should(QueryBuilders.matchQuery("subtitle", search)).boost(5)
//                    .should(QueryBuilders.matchPhraseQuery("content", search)).boost(5)
//            );
//            highlightBuilder.field("title");
//            highlightBuilder.field("subtitle");
//            highlightBuilder.field("author");
//            highlightBuilder.field("keyword");
//            highlightBuilder.field("content");
//            build.highlighter(highlightBuilder);
//        }
//        if (StringUtils.isNotEmpty(dwdEsVO.getFieldName())) {
//            boolQueryBuilder.must(QueryBuilders.matchPhraseQuery("field_name", dwdEsVO.getFieldName()));
//        }
//        if (dwdEsVO.getContentType() != null) {
//            boolQueryBuilder.must(QueryBuilders.termQuery("content_type", dwdEsVO.getContentType()));
//        }
//        if (dwdEsVO.getNoId() != null) {
//            boolQueryBuilder.mustNot(QueryBuilders.termQuery("id", dwdEsVO.getNoId()));
//        }
//        if (StringUtils.isEmpty(search)) {
//            build.sort("read_num", SortOrder.DESC);
//        }
//        //分页查询
//        build.from(dwdEsVO.getPageNo()).size(dwdEsVO.getPageSize());
//        build.query(boolQueryBuilder);
//        IPage<AriNews> dwdTechInfoIPage = esUtils.searchPage(index_ari_news, build, AriNews.class);
//
//        String hostIp = nginxUrl.substring(nginxUrl.indexOf("//") + 2, nginxUrl.lastIndexOf(":"));
//        dwdTechInfoIPage.getRecords().stream().forEach(one -> {
//            one.setContent("");
//            //TODO 后期可以注释，为了保证拿到现场的图片数据正常访问
//            if (StringUtils.isNotEmpty(one.getCover())) {
//                one.setCover(one.getCover().replaceAll(Constants.INIT_IP, hostIp));
//            }
//            if (StringUtils.isNotEmpty(one.getExtras())) {
//                one.setExtras(one.getExtras().replaceAll(Constants.INIT_IP, hostIp));
//            }
//        });
//        return dwdTechInfoIPage;
//    }

    @Override
    public void mysqlToEs(String indexName) {
        List<AriMysqlEs> dwdMysqlEsList = null; // status = 0
        for (AriMysqlEs dwdMysqlEs : dwdMysqlEsList) {
            try {
                //拿到es索引
                String index = dwdMysqlEs.getIndex();
                if (StringUtils.isEmpty(index)) {
                    continue;
                }
                boolean indexExists = true;
                // 查看是否有索引存在
                if (!esUtils.testExist(index)) {
                    indexExists = false;
                }
                String lastField = dwdMysqlEs.getLastField();
                //给个初始时间
                String lastTime = INIT_TIME;
                if (dwdMysqlEs.getLastTime() != null) {
                    lastTime = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(dwdMysqlEs.getLastTime());
                }
                //获取当前索引的id
                String esId = dwdMysqlEs.getEsId();
                //拿到查询语句
                String selectSql = dwdMysqlEs.getSelectSql();
                int current = 0;
                int size = ES_COUNT;

                // 循环操作进行一个 300 数据同步
                while (true) {
                    String sql = selectSql;
                    if (!StringUtils.isEmpty(lastField)) {
                        if (sql.contains("where") || sql.contains("WHERE")) {
                            sql = sql + " and " + lastField + " >= '" + lastTime + "'";
                        } else {
                            sql = sql + " where " + lastField + " >= '" + lastTime + "'";
                        }
                    }
                    sql = sql + " limit " + current + " , " + size;

                    // 拼接sql select * from ari_news where is_deleted=0 and update_time > 上一次同步的时间

                    List<HashMap<String, Object>> data = null;//dwdMysqlEsMapper.getDataBySql(sql);
                    //拿到请求
                    BulkRequest bulkRequest = new BulkRequest();
                    if (!CollectionUtils.isEmpty(data)) {
                        for (HashMap<String, Object> datum : data) {
                            try {
                                if (!indexExists) {
                                    Long initId = -1L;
                                    Map<String, Object> createMap = new HashMap<>();
                                    for (String key : datum.keySet()) {

                                        if ("id".equals(key)) {
                                            createMap.put(key, initId);
                                        } else if ("read_num".equals(key)) {
                                            createMap.put(key, 0);
                                        } else {

                                            // mysql 字段数据全盘复制

                                            createMap.put(key, "");
                                        }
                                    }
//                                    BulkRequest bulkRequestNew = new BulkRequest();
//                                    IndexRequest indexRequestNew = new IndexRequest(index);
//                                    indexRequestNew.id(String.valueOf(initId));
//                                    indexRequestNew.source(JSONObject.toJSONString(createMap), XContentType.JSON);
//                                    bulkRequestNew.add(indexRequestNew);
//                                    bulkRequestNew.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
//                                    esUtils.bulkData(bulkRequestNew);
//                                    esUtils.deleteData(index, String.valueOf(initId));
                                    indexExists = true;
                                }

                                // 进行入库

//                                IndexRequest indexRequest = new IndexRequest(index);
//                                String id =  datum.get(esId).toString();
//                                indexRequest.id(id);
//                                String s = JSONObject.toJSONString(datum, SerializerFeature.WriteMapNullValue);
//                                indexRequest.source(s, XContentType.JSON);
//                                bulkRequest.add(indexRequest);
                            } catch (Exception e) {
                                log.error("添加数据失败", e);
                            }
                        }
                        bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
                        // esUtils.bulkData(bulkRequest);
                        // 进行下次查询的一个操作
                        current = current + size;
                    } else {
                        break;
                    }
                }


                if (!StringUtils.isEmpty(lastField)) {
                    // 同步解耦进行一个时间修改
//                    dwdMysqlEsMapper.update(null, new LambdaUpdateWrapper<AriMysqlEs>()
//                            .eq(AriMysqlEs::getId, dwdMysqlEs.getId())
//                            .set(AriMysqlEs::getLastTime, new Date())
//                    );
                }
                log.info("索引：" + index + "同步完成");
            } catch (Exception e) {
                log.error("索引同步失败：" + dwdMysqlEs.getIndex(), e);
            }
        }
    }



}




